package org.example.wordCount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * @author huangqihan
 * @date 2021/2/9
 */

public class WordCountByBatch {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        String path = "C:\\Users\\huangqihanssh\\Desktop\\apexsoft\\omp\\quartz-example\\src\\main\\resources\\testData.txt";
        DataSource<String> stringDataSource = env.readTextFile(path);

        AggregateOperator<Tuple2<String, Integer>> sum = stringDataSource.flatMap(new MyFlatMapper())
                // group by index 0
                .groupBy(0)
                // sum by index 1
                .sum(1);
        sum.print();
    }

    public static class MyFlatMapper implements FlatMapFunction<String, Tuple2<String, Integer>> {

        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            String[] words = value.split(" ");
            Arrays.stream(words).forEach((word) -> {
                out.collect(new Tuple2<>(word, 1));
            });
        }
    }
}
